This PR introduces an implementation of result storage using Apache Iceberg.

How to enable the Iceberg result storage

Go to storage-config.yaml,

• change result-storage-mode to iceberg
• configure storage.iceberg.catalog.jdbc section,

iceberg:
    catalog:
      jdbc: # currently we only support storing catalog info via jdbc, i.e. https://iceberg.apache.org/docs/1.7.1/jdbc/
        url: "jdbc:mysql://localhost:3306/texera_iceberg?serverTimezone=UTC"
        username: ""
        password: ""

make sure the JDBC is accessible via the url, username, and password

Major changes

• Introduced IcebergDocument: a thread-safe implementation of VirtualDocument for storing and reading results in Iceberg tables.
• Introduced IcebergTableWriter: an append-only writer for Iceberg tables with configurable buffer size.
• Added support for new configuration properties under storage.iceberg to specify catalog and table settings.

Introduced Dependencies

In workflow-core, some new packages are added

• Iceberg-related packages
• Hadoop common. The reason of adding this dependency is to pass the compilation: In the source code of iceberg-parquet, the line 160,
  although the file is not of type HadoopOutputFile, it still creats a Hadoop Configuration() as the placeholder. During the runtime, we don't have any dependency on Hadoop or HDFS.

Overview of the behavior IcebergDocument and IcebergWriter

• IcebergDocument:

  • Handles reading and managing data in Iceberg tables.
  • Initializes the table during construction, creating it if it does not exist or overriding it if specified.
  • Supports iterator-based incremental read operations.
  • Thread-safe for read and clear operations.

• IcebergTableWriter:

  • Writes data to Iceberg tables in an append-only manner.
  • Creates new Parquet files for every buffer flush, ensuring immutability.
  • Not thread-safe, so it should only be accessed by one thread at a time.

How the result will be stored via Iceberg tables

• Given a storage key, a table named key will be created.
• To append tuples to the table key, each worker will append immutable parquet files to the table's data space using IcebergTableWriter. To avoid the parquet filename collision, each worker will prefix its created file with ${workerIndex}_${fileIndex}, in which workerIndex is its index, and fileIndex is a number maintained that increased by 1 every time a new data file is created and flushed by the writer.
• To read the tuples, the reader uses the iterator returned by IcebergDocument.get. This iterator can incrementally read new data while writers are appending tuples.